package org.zhaya.sdk.processors;

import io.logz.sawmill.Doc;
import io.logz.sawmill.ProcessResult;
import io.logz.sawmill.Processor;
import io.logz.sawmill.annotations.ProcessorProvider;
import io.logz.sawmill.exceptions.ProcessorConfigurationException;
import io.logz.sawmill.utilities.JsonUtils;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.math.NumberUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;

import static java.util.Objects.requireNonNull;

@Slf4j
@ProcessorProvider(type = "numberExchange", factory = NumberExchangeProcessor.Factory.class)
public class NumberExchangeProcessor implements Processor {
    private final List<String> fields;
    private final List<String> targetFields;
    private final int inRadix;
    private final int outRadix;

    public NumberExchangeProcessor(List<String> fields, List<String> targetFields, int inRadix, int outRadix) {
        this.fields = requireNonNull(fields, "fields cannot be null");
        this.targetFields = targetFields;
        this.inRadix = inRadix;
        this.outRadix = outRadix;
    }

    @Override
    public ProcessResult process(Doc doc) {
        List<String> missingFields = new ArrayList<>();
        List<String> missingTargetFields = new ArrayList<>();

        for (int i = 0; i < fields.size(); i++) {
            if (!doc.hasField(fields.get(i))) {
                missingFields.add(fields.get(i));
                continue;
            }

            if (CollectionUtils.isEmpty(targetFields)) {
                try {
                    doc.addField(fields.get(i), convertByRadix(doc.getField(fields.get(i)), inRadix, outRadix));
                } catch (Exception e) {
                    return ProcessResult.failure(String.format("Could not convert value, field is [%s], msg=[%s]",
                            fields.get(i), e.getMessage()));
                }
            } else {
                if (targetFields.size() != fields.size()) {
                    return ProcessResult.failure("The number of fields and targetFields need to be corresponding and consistent");
                }
                if (!doc.hasField(targetFields.get(i))) {
                    missingTargetFields.add(targetFields.get(i));
                    continue;
                }
                try {
                    doc.addField(targetFields.get(i), convertByRadix(doc.getField(fields.get(i)), inRadix, outRadix));
                } catch (Exception e) {
                    return ProcessResult.failure(String.format("Could not convert value, field is [%s], msg=[%s]",
                            fields.get(i), e.getMessage()));
                }
            }
        }

        if (CollectionUtils.isNotEmpty(missingFields)) {
            return ProcessResult.failure(String.format("failed to convert fields in path %s, fields are missing or not instance of [String]", missingFields));
        }
        if (CollectionUtils.isNotEmpty(missingTargetFields)) {
            return ProcessResult.failure(String.format("failed to convert fields in path %s, targetFields are missing or not instance of [String]", missingTargetFields));
        }

        return ProcessResult.success();
    }

    public static class Factory implements Processor.Factory {

        @Override
        public NumberExchangeProcessor create(Map<String, Object> config) {
            Configuration sysConvertConfig = JsonUtils.fromJsonMap(Configuration.class, config);

            if (CollectionUtils.isEmpty(sysConvertConfig.getFields())) {
                throw new ProcessorConfigurationException("failed to parse sysConvert config, cannot be processed without fields");
            }

            if (errorRadix(sysConvertConfig.getIn_radix()) || errorRadix(sysConvertConfig.getOut_radix())) {
                throw new ProcessorConfigurationException("failed to parse sysConvert config, cannot be processed without legal in_radix/out_radix, it should be in the (2,36] range");
            }

            return new NumberExchangeProcessor(sysConvertConfig.getFields(),
                    sysConvertConfig.getTargetFields(), sysConvertConfig.getIn_radix(), sysConvertConfig.getOut_radix());
        }

        private static boolean errorRadix(int num) {
            return num < Character.MIN_RADIX || num > Character.MAX_RADIX;
        }
    }

    @Data
    @Builder
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Configuration implements Processor.Configuration {
        private List<String> fields;
        private List<String> targetFields;
        private int in_radix;
        private int out_radix;

    }

    private static Object convertByRadix(Object value, int inRadix, int outRadix) throws Exception {
        if (Objects.isNull(value)) {
            throw new Exception("The value cannot null!");
        }

        try {
            //转化为10进制数字
            final long number10 = Long.parseLong(String.valueOf(value), inRadix);
            //转化为指定进制字符串
            final String resultStr = Long.toString(number10, outRadix);
            return value instanceof Number && outRadix <= 10 ? NumberUtils.toLong(resultStr) : resultStr;
        } catch (Exception e) {
            log.error("convertByRadix error:{},{},{}", value, inRadix, outRadix, e);
            throw new Exception("The in_radix or out_radix is not up to standard. The error is : " + e);
        }

    }

}